package com.o2o.cleaning.month.platform.ebusiness_plat.jumei_2019_7

//import brand.UpdataTable_new.brand_join_res
import java.text.SimpleDateFormat

import com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.brand_join_res
import org.apache.spark.sql.SparkSession
//import platform.checkData.CheckData
//import platform.jumei.jumei_th1.Jumei_utils.path_util

/**
  * 聚美代码最新版
  */

//每个月处理之前提取上个月的【也可以当月数据处理完毕之后提取】
//val isOnsell_false_Url = "o2o-dataproces-group/liu_taowei/month_data/plat_need/jumei/is_onsell_false/*"
//这个是聚美的分类表路径
//  地址表路径
//  val feiziying_addressUrl ="s3a://o2o-dataproces-group/liu_taowei/month_data/address/jumei_address/jumei_address_20190508/"

object JuMei {

  /** *******************重要重要重要重要重要重要 ************************************************************/
  val platformName = "jumei"
  val year = 2021
  val month = 6
//  val sart = s"2020-6-1 00:00:00"
  val sart = s"2021-${month}-1 00:00:00"
//  val end =  s"2020-6-15 24:00:00"
  val end =  s"2021-${month}-30 00:00:00"
  val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
  //  促销商品的【开始】时间戳
  val start_time = simpleDateFormat.parse(sart).getTime()/1000
  //  促销商品的【结束】时间戳
  val end_time  = simpleDateFormat.parse(end).getTime()/1000

  val timeStamp = "1624982400"


// 当月的原始数据路径
//  val sourceData_url =  s"s3a://s3a://o2o-sourcedata/obs-source-2020/618/gov/jumei"
  val sourceData_url =  s"s3a://o2o-dataproces-group/zyf/${year}/${month}/jumei/"
//  分类表路径
//  val jumeiCate_Url = path_util.newnest_path("s3a://o2o-dimension-table/category_table/categoryFile_tao/jumei_2019_9")
  val jumeiCate_Url = "s3a://o2o-dimension-table/category_table/categoryFile_tao/jumei_2019_9"
//  结果文件路径
  val result_goodUrl = s"s3a://o2o-dataproces-group/zyf/${year}/${month}/${platformName}_result/"
  //o2o-dataproces-group/li_yinchao/lyc/platform/jumei/2020/4/brand_join_text_2/

//  val result_goodUrl = s"s3a://o2o-dataproces-group/li_yinchao/lyc/platform/${platformName}/${year}/${month}/brandJoin_1/"
  //    老品牌表路径
  val brandTable_old = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/${platformName}/"
  //    品牌表结果路径
  val brandTable_Result = s"s3a://o2o-dimension-table/brandName_table/${year}/${month+1}/${platformName}/"
  //    新增品牌路径
  val brandTable_addNew = s"s3a://o2o-dimension-table/brandName_table/newAddBrand/${year}/${month}/${platformName}/"
  /** *******************重要重要重要重要重要重要 *************************************************************/



  def main(args: Array[String]): Unit = {

   val spark = SparkSession.builder()
      .master("local[*]")
      .appName("JuMei")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()
    val sc = spark.sparkContext
    val sqlContext = spark.sqlContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")

//    读取原始数据路径
    val rdd = spark.read.json(sourceData_url)
//   这个是对原始数据进行清洗
    val frame = Jumei_utils.base_util.base_util_jumei(rdd,spark)
      .where("sellCount > 0 and priceText > 0 and sellCount is not null and priceText is not null")
//      2019——6月数据处理变动
      .where("size(add_to_field) > 0")

    /*** 聚美 价格数据清洗结果输出============================================================================================= */
    println("=========================聚美 数据解析 过滤==================================================================================")
    val frame1 = Jumei_utils.base_util.filter_isNotOnSell(frame,spark,start_time.toString,end_time.toString)

    println("=======================分类表 有歧义数据查找====================================================================================")
    /**分类*/
    val jumei_xiuzheng_data = Jumei_utils.cate_util.jumei_cate_util(frame1,spark)
      .dropDuplicates("good_id")

    println("=======================聚美 地址表关联================================================================================")

    val source_data = Jumei_utils.address_util.is_self_data(jumei_xiuzheng_data,timeStamp)
         .dropDuplicates("good_id")
    //println(source_data.count())
    //source_data.show(10)


    /**
      * @param sourceDataPath:DataFrame  跑完分类的【数据】
      * @param sourceResultPath：String  跑完品牌的数据结果路径
      * @param year：int                 年
      * @param month：int                月
      * @param platformName ：string     平台名称
      * @param spark
      */
    val brand = new brand_join_res
    //    读取老品牌表路径
    //val brandTableOld = spark.read.json(brandTable_old)
    brand.brandJoinResult(source_data,result_goodUrl,year,month,platformName,spark)


    sc.stop()

  }
}
